package com.gaofans.nrpc.client.handler;

import com.gaofans.nrpc.common.context.ContextRequest;
import com.gaofans.nrpc.common.filter.DefaultFilterChain;
import com.gaofans.nrpc.common.filter.FilterChain;
import com.gaofans.nrpc.common.processor.RpcContext;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;

/**
 * request出站过滤器
 * @author GaoFans
 */
public class RequestOutFilterHandler extends ChannelOutboundHandlerAdapter {

    private final List<ClientFilter> filters;
    private final static Logger LOGGER = LoggerFactory.getLogger(RequestOutFilterHandler.class);

    public RequestOutFilterHandler(List<ClientFilter> filters) {
        this.filters = Objects.requireNonNull(filters);
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
        if (msg instanceof ContextRequest && filters.size() > 0){
            ContextRequest request = (ContextRequest) msg;
            RpcContext context = new RpcContext(ctx, request.getSequenceId());
            request.executor().submit(() -> {
                try {
                    FilterChain<ContextRequest> chain = new DefaultFilterChain<>(filters, context);
                    chain.doFilter((ContextRequest) msg)
                            .doOnSuccess(unused -> ctx.writeAndFlush(msg))
                            .doOnError(throwable -> LOGGER.error("过滤器链调用异常",throwable))
                            .doFinally(signalType -> context.clear())
                            .subscribe();
                }catch (Exception e){
                    context.clear();
                }
            });

        }else {
            ctx.writeAndFlush(msg);
        }
    }
}
